package com.atgugu.realtime.app.dws;

import com.atgugu.realtime.app.BaseSqlApp;
import com.atgugu.realtime.bean.KeywordStats;
import com.atgugu.realtime.common.Constant;
import com.atgugu.realtime.function.IkAnalyzer;
import com.atgugu.realtime.function.KwProduct;
import com.atgugu.realtime.util.FlinkSinkUtil;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.ZoneOffset;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2022/1/5 9:12
 */
public class DwsProductKeywordStatsApp extends BaseSqlApp {
    public static void main(String[] args) {
        new DwsProductKeywordStatsApp().init(4005, 1, "DwsProductKeywordStatsApp");
        
    }
    
    @Override
    protected void run(StreamTableEnvironment tEnv) {
        tEnv.getConfig().setLocalTimeZone(ZoneOffset.ofHours(8));
        // 1. 建立动态表, 从kakfa读数据: dwd_page
        tEnv.executeSql("create table product_stats(" +
                            "   stt string, " +
                            "   edt string, " +
                            "   sku_name string, " +
                            "   click_ct bigint, " +
                            "   order_ct bigint, " +
                            "   cart_ct bigint " +
                            ")with(" +
                            "   'connector'='kafka', " +
                            "   'properties.bootstrap.servers'='" + Constant.KAFKA_BROKERS + "', " +
                            "   'properties.group.id'='DwsProductKeywordStatsApp', " +
                            "   'topic'='" + Constant.TOPIC_DWS_PRODUCT_STATS + "', " +
                            "   'format'='json', " +
                            "   'scan.startup.mode'='latest-offset' " +
                            ")");
        
        // 1. 过滤出需要的数据: 三个ct至少一个不为0
        Table t1 = tEnv.sqlQuery("select " +
                                     "* " +
                                     "from " +
                                     "product_stats " +
                                     "where click_ct > 0 " +
                                     "or order_ct > 0 " +
                                     "or cart_ct > 0");
        tEnv.createTemporaryView("t1", t1);
        
        // 2. 分词
        tEnv.createTemporaryFunction("ik_analyzer", IkAnalyzer.class);
        Table t2 = tEnv.sqlQuery("select" +
                                     " stt, " +
                                     " edt, " +
                                     " keyword, " +
                                     " click_ct , " +
                                     " order_ct , " +
                                     " cart_ct  " +
                                     "from t1 " +
                                     "join lateral table(ik_analyzer(sku_name)) on true");
        tEnv.createTemporaryView("t2", t2);
        
        // 3. 使用表值函数, 列转行
        tEnv.createTemporaryFunction("kw_product", KwProduct.class);
        
        Table t3 = tEnv.sqlQuery("select" +
                                     " stt," +
                                     " edt, " +
                                     " keyword, " +
                                     " source, " +
                                     " ct " +
                                     "from t2 " +
                                     "join lateral table(kw_product(click_ct, order_ct, cart_ct)) on true");
        tEnv.createTemporaryView("t3", t3);
        
        // 4. 聚合: 按照stt,edt,keyword, source 聚合
        Table result = tEnv.sqlQuery("select" +
                                         " stt," +
                                         " edt, " +
                                         " keyword, " +
                                         " source, " +
                                         " sum(ct) ct, " +
                                         " unix_timestamp() * 1000 ts " +
                                         "from t3 " +
                                         "group by " +
                                         " stt," +
                                         " edt, " +
                                         " keyword, " +
                                         " source ");
        
        tEnv.toRetractStream(result, KeywordStats.class)
            .filter(t -> t.f0)
            .map(t -> t.f1)
            .addSink(FlinkSinkUtil.getClickHouseSink("gmall2021", "keyword_stats_2021", KeywordStats.class));
        
    }
}
/*
现有:
sku             click_ct  order_ct  cart_ct
小米手机        10          0           20
小米电视        20          0           20
--------
第一步:
小米       10          0          20
手机        10         0           20

----
制表函数

----
目的:
小米   "click"   10
小米   "order"   0
小米   "cart"   2

手机   "click"   10
手机   "order"   0
手机   "cart"   2


小米    "click"   20
 */